package task4

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment}

/**
 * @author jhhe66
 * @date 2021/6/18 11:26
 */
object State {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val kvDS: KeyedStream[(String, Int), String] = env.socketTextStream("centos001", 9999)
      .map { line =>
        val arr = line.split(",")
        (arr(0).trim, arr(1).trim.toInt)
      }.keyBy(_._1)

    val res = kvDS.map(
      new RichMapFunction[(String, Int), (String, Int)] {
        var maxValueState: ValueState[Int] = _

        override def open(parameters: Configuration): Unit = {
          val maxValue: ValueStateDescriptor[Int] = new ValueStateDescriptor[Int]("maxValue", classOf[Int])
          maxValueState = getRuntimeContext.getState(maxValue)
        }

        override def map(in: (String, Int)): (String, Int) = {
          //将较大的数更新进状态
          maxValueState.update(Math.max(maxValueState.value(), in._2))
          //返回对应的key的最大值
          (in._1, maxValueState.value())
        }
      }
    )
    res.print()
    env.execute()
  }
}
